package com.utopa.canal.client.handler;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.utopa.canal.client.annotation.CanalListener;
import com.utopa.canal.client.config.CanalApplicationEvent;
import com.utopa.canal.client.listener.CanalMessageListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.*;

/**
 * @author authorZhao
 * @date 2020年04月15日
 */
@Slf4j
public class CanalMessageListenerProxy implements InvocationHandler {
    private CanalListener canalListener;
    private Object bean;
    private Method method;

    /**
     * 采用构造注入，通过反射执行方法
     */
    public CanalMessageListenerProxy(Object bean, Method method, CanalListener canalListener) {
        this.bean = bean;
        this.method = method;
        this.canalListener = canalListener;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] params) throws Throwable {
        if (Object.class.equals(method.getDeclaringClass())) {
            return method.invoke(this, params);
        }
        CanalEntry.Entry entry = (CanalEntry.Entry) params[0];
        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
            return null;
        }
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        Class<?> clazz = this.method.getParameterTypes()[0];
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            CanalEntry.EventType eventType = rowChange.getEventType();
            RowDataHandler bean = null;
            Class<? extends RowDataHandler> handler = canalListener.handler();
            if (handler != null) {
                bean = CanalApplicationEvent.getBean(handler);
            }
            if (bean == null) {
                bean = CanalApplicationEvent.getBean(DefaultRowDataHandler.class);
            }
            Object targetObj = bean.changeType(rowData, clazz, eventType);
            return this.method.invoke(this.bean, targetObj, eventType);
        }
        return null;
    }

    /**
     * @param bean
     * @param method
     * @param canalListener
     * @return
     */
    public static CanalMessageListener createProxyObject(Object bean, Method method, CanalListener canalListener) {
        CanalMessageListenerProxy invocationHandler = new CanalMessageListenerProxy(bean, method, canalListener);
        ClassLoader classLoader = bean.getClass().getClassLoader();
        Class[] interfaces = {CanalMessageListener.class};
        // 只为 CanalMessageListener 类生成代理对象
        CanalMessageListener canalMessageListener = (CanalMessageListener) Proxy.newProxyInstance(classLoader, interfaces, invocationHandler);
        return canalMessageListener;
    }
}
